package com.zhitan.ems.app.sse;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;

import com.zhitan.ems.app.iot.entity.DeviceTopologyEntity;
import com.zhitan.ems.app.iot.service.DeviceTopologyService;
import com.zhitan.ems.app.iot.vo.TopologyPointDataVO;
import com.zhitan.ems.common.core.redis.RedisCache;

import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpStatus;
import lombok.extern.slf4j.Slf4j;

/**
 * SSE连接管理服务
 * 
 * @author zhitan
 */
@Slf4j
@Service
public class SseTopologyServiceImpl {
    /**
     * 容器，保存连接，用于输出返回
     */
    private static final Map<String, SseEmitterUTF8> SSE_CACHE = MapUtil.newConcurrentHashMap();
    
    /**
     * 连接最后活跃时间缓存
     */
    private static final Map<String, Long> LAST_ACTIVE_TIME = new ConcurrentHashMap<>();
    
    private static final Map<String, TopologyPointDataVO> TOPOLOGYMAP = new ConcurrentHashMap<>();
    
    /**
     * 重试次数
     */
    private final Integer RESET_COUNT = 3;

    /**
     * 重试等待时间 单位 ms
     */
    private final Integer RESET_TIME = 5000;
    
    /** 连接超时时间10分钟 */
    private final long TIME_OUT = 10 * 60 * 1000L;
    
    /** 心跳间隔时间 30秒 */
    private final long HEARTBEAT_INTERVAL = 30 * 1000L;
    
    /** 连接检查间隔时间 1分钟 */
    private final long CONNECTION_CHECK_INTERVAL = 60 * 1000L;
    
    @Autowired
    private DeviceTopologyService deviceTopologyService;
    @Resource
    private RedisCache redisCache;
    /**
     * 心跳和连接检查调度器
     */
    private ScheduledExecutorService scheduler;

    /**
     * 初始化调度器
     */
    @PostConstruct
    public void init() {
        scheduler = new ScheduledThreadPoolExecutor(2, r -> {
            Thread t = new Thread(r, "sse-heartbeat-thread");
            t.setDaemon(true);
            return t;
        });
        
        // 启动心跳任务
        scheduler.scheduleAtFixedRate(this::sendHeartbeats, 
            HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
            
        // 启动连接检查任务
        scheduler.scheduleAtFixedRate(this::checkConnections, 
            CONNECTION_CHECK_INTERVAL, CONNECTION_CHECK_INTERVAL, TimeUnit.MILLISECONDS);
            
        log.info("SSE连接管理服务初始化完成");
    }

    /**
     * 销毁调度器
     */
    @PreDestroy
    public void destroy() {
        if (scheduler != null && !scheduler.isShutdown()) {
            scheduler.shutdown();
            try {
                if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        log.info("SSE连接管理服务已销毁");
    }

    /**
     * 创建连接
     * 
     * @param topologyId 拓扑ID
     * @return SseEmitterUTF8 SSE连接对象
     */
    public SseEmitterUTF8 createConnect(String topologyId) {
        // 检查参数
        if (StrUtil.isBlank(topologyId)) {
            log.warn("SSE注册信息不全: topologyId为空");
            return null;
        }
        
        // 检查设备是否存在（暂时注释掉以便测试）
        DeviceTopologyEntity topology = deviceTopologyService.getById(topologyId);
        if (null == topology) {
            log.info("SSE注册设备信息不匹配: {}", topologyId);
            return null;
        }
        // 设置超时时间 单位毫秒，0表示不过期
        SseEmitterUTF8 sseEmitter = new SseEmitterUTF8(TIME_OUT);
        // 如果存在上一个，关闭连接
        if (SSE_CACHE.containsKey(topologyId)) {
            closeConnect(topologyId);
        } else {
            // 集群下，广播通知关闭
            ClientNotifyVo resp = new ClientNotifyVo(ClientOptType.CLOSE_CONNECT, sseEmitter.getUuid());
            sendMessageByPublish(topologyId, resp, true);
        }
        // 注册回调
        // 长链接完成后回调接口(即关闭连接时调用)
        sseEmitter.onCompletion(completionCallBack(topologyId));
        // 连接超时回调
        sseEmitter.onTimeout(timeoutCallBack(topologyId));
        // 推送消息异常时，回调方法
        sseEmitter.onError(errorCallBack(topologyId));

        SSE_CACHE.put(topologyId, sseEmitter);
        LAST_ACTIVE_TIME.put(topologyId, System.currentTimeMillis());
        log.info("创建新的SSE连接，拓扑ID：{}，连接UUID：{}，累计连接数：{}", 
            topologyId, sseEmitter.getUuid(), SSE_CACHE.size());
        // 发送连接成功通知
        ClientNotifyVo connectSuccess = new ClientNotifyVo(ClientOptType.CONNECT_SUCCESS, "连接建立成功");
        sendMsgToClientByTopology(topologyId, connectSuccess);
        return sseEmitter;
    }

    /**
     * 关闭连接
     *
     * @param topologyId 拓扑ID
     */
    public void closeConnectTopology(String topologyId) {
        if (StrUtil.isBlank(topologyId)) {
            return;
        }
        
        DeviceTopologyEntity topology = deviceTopologyService.getById(topologyId);
        if (null == topology) {
            return;
        }
        closeConnect(topologyId);
    }

    /**
     * 内部关闭连接方法
     *
     * @param topologyId 拓扑ID
     */
    private void closeConnect(String topologyId) {
        SseEmitterUTF8 sseEmitter = SSE_CACHE.get(topologyId);
        if (sseEmitter != null) {
            log.info("关闭SSE连接，拓扑ID：{}，连接UUID：{}", topologyId, sseEmitter.getUuid());
            sseEmitter.complete();
            removeUser(topologyId);
        }
    }

    /**
     * 发送消息
     * 
     * @param topologyId 拓扑ID
     * @param resp 消息内容
     * @param broadcast 是否广播
     */
    public void sendMessageByPublish(String topologyId, ClientNotifyVo resp, boolean broadcast) {
        if (StrUtil.isBlank(topologyId) || resp == null) {
            return;
        }
        
        if (SSE_CACHE.containsKey(topologyId)) {
            if (resp.getOptType() == ClientOptType.CLOSE_CONNECT) {
                // 防止自己清除自己
                SseEmitterUTF8 sseEmitter = SSE_CACHE.get(topologyId);
                if (sseEmitter.getUuid().equals(String.valueOf(resp.getData()))) {
                    return;
                }
                log.info("SSE关闭上一个连接，拓扑ID：{}", topologyId);
                closeConnect(topologyId);
                return;
            }
            sendMsgToClientByTopology(topologyId, resp);
            return;
        }
        if (broadcast) {
            // TODO: 实现集群广播逻辑
        }
    }

    /**
     * 推送消息到客户端
     *
     * @param topologyId 拓扑ID
     * @param msg 推送信息
     **/
    public void sendMsgToClientByTopology(String topologyId, ClientNotifyVo msg) {
        if (StrUtil.isBlank(topologyId) || msg == null) {
            return;
        }
        
        SseEmitterUTF8 emitter = SSE_CACHE.get(topologyId);
        if (emitter == null) {
            log.warn("SSE连接不存在，拓扑ID：{}", topologyId);
            return;
        }
        
        // 更新最后活跃时间
        LAST_ACTIVE_TIME.put(topologyId, System.currentTimeMillis());
        
        log.debug("SSE推送消息：拓扑ID={}，消息类型={}", topologyId, msg.getOptType());
        SseEmitterUTF8.SseEventBuilder sendData = SseEmitterUTF8.event().id(String.valueOf(HttpStatus.HTTP_OK))
                .data(msg, MediaType.APPLICATION_JSON);
        
        // 异步发送消息，避免阻塞主线程
        CompletableFuture.runAsync(() -> {
            try {
                emitter.send(sendData);
            } catch (Exception e) {
                // 推送消息失败，记录错误日志，进行重推
                log.error("推送消息失败，拓扑ID：{}，消息：{}，尝试进行重推", topologyId, msg, e);
                retrySendMessage(topologyId, sendData);
            }
        });
    }
    
    /**
     * 重试发送消息
     * 
     * @param topologyId 拓扑ID
     * @param sendData 消息数据
     */
    private void retrySendMessage(String topologyId, SseEmitterUTF8.SseEventBuilder sendData) {
        // 推送消息失败后，每隔5s推送一次，推送3次
        for (int i = 0; i < RESET_COUNT; i++) {
            try {
                // 防止等待过程中，连接断开
                Thread.sleep(RESET_TIME);
                SseEmitterUTF8 emitter = SSE_CACHE.get(topologyId);
                if (emitter == null) {
                    log.error("{}的第{}次消息重推失败，未创建长链接", topologyId, i + 1);
                    continue;
                }
                emitter.send(sendData);
                log.info("{}的第{}次消息重推成功", topologyId, i + 1);
                // 更新最后活跃时间
                LAST_ACTIVE_TIME.put(topologyId, System.currentTimeMillis());
                return;
            } catch (Exception ex) {
                log.error("{}的第{}次消息重推失败", topologyId, i + 1, ex);
                continue;
            }
        }
        log.error("{}的消息重推全部失败", topologyId);
    }

    /**
     * 长链接完成后回调接口(即关闭连接时调用)
     *
     * @param topologyId 拓扑ID
     **/
    private Runnable completionCallBack(String topologyId) {
        return () -> {
            log.info("SSE连接正常结束：{}", topologyId);
            removeUser(topologyId);
        };
    }

    /**
     * 连接超时时调用
     *
     * @param topologyId 拓扑ID
     **/
    private Runnable timeoutCallBack(String topologyId) {
        return () -> {
            log.info("SSE连接超时：{}", topologyId);
            removeUser(topologyId);
        };
    }

    /**
     * 推送消息异常时，回调方法
     *
     * @param topologyId 拓扑ID
     **/
    private Consumer<Throwable> errorCallBack(String topologyId) {
        return throwable -> {
            log.error("SSE连接异常，拓扑ID：{}", topologyId, throwable);
            // 移除异常连接
            removeUser(topologyId);
        };
    }

    /**
     * 移除用户连接
     *
     * @param topologyId 拓扑ID
     **/
    private void removeUser(String topologyId) {
        if (StrUtil.isNotBlank(topologyId)) {
            SSE_CACHE.remove(topologyId);
            LAST_ACTIVE_TIME.remove(topologyId);
            log.info("移除SSE连接，拓扑ID：{}，剩余连接数：{}", topologyId, SSE_CACHE.size());
        }
    }
    
    /**
     * 获取当前连接数
     * 
     * @return 连接数
     */
    public int getConnectionCount() {
        return SSE_CACHE.size();
    }
    
    /**
     * 发送心跳消息
     * 
     * @param topologyId 拓扑ID
     */
    public void sendHeartbeat(String topologyId) {
        if (StrUtil.isBlank(topologyId)) {
            return;
        }
        
        ClientNotifyVo pingMsg = new ClientNotifyVo(ClientOptType.PING, System.currentTimeMillis());
        sendMsgToClientByTopology(topologyId, pingMsg);
    }
    
    /**
     * 批量发送心跳消息
     */
    private void sendHeartbeats() {
        try {
            for (String topologyId : SSE_CACHE.keySet()) {
                sendHeartbeat(topologyId);
            }
        } catch (Exception e) {
            log.error("批量发送心跳消息异常", e);
        }
    }
    
    /**
     * 检查并清理过期连接
     */
    private void checkConnections() {
        try {
            long currentTime = System.currentTimeMillis();
            for (Map.Entry<String, Long> entry : LAST_ACTIVE_TIME.entrySet()) {
                String topologyId = entry.getKey();
                Long lastActiveTime = entry.getValue();
                // 如果超过2倍心跳间隔时间没有活跃，则认为连接可能已断开
                if (currentTime - lastActiveTime > HEARTBEAT_INTERVAL * 2) {
                    log.warn("检测到可能已断开的SSE连接，拓扑ID：{}，最后活跃时间：{}", topologyId, lastActiveTime);
                    SseEmitterUTF8 emitter = SSE_CACHE.get(topologyId);
                    if (emitter != null) {
                        // 尝试发送心跳确认连接状态
                        ClientNotifyVo checkMsg = new ClientNotifyVo(ClientOptType.PING, "connection_check");
                        try {
                            emitter.send(SseEmitterUTF8.event().data(checkMsg, MediaType.APPLICATION_JSON));
                            // 更新活跃时间
                            LAST_ACTIVE_TIME.put(topologyId, currentTime);
                        } catch (Exception e) {
                            log.info("确认连接已断开，关闭SSE连接，拓扑ID：{}", topologyId);
                            closeConnect(topologyId);
                        }
                    } else {
                        // 清理无效记录
                        removeUser(topologyId);
                    }
                }
            }
        } catch (Exception e) {
            log.error("检查SSE连接状态异常", e);
        }
    }
    
    /**
     * 获取连接状态信息
     * 
     * @return 连接状态信息
     */
    public Map<String, Object> getConnectionStats() {
        Map<String, Object> stats = new ConcurrentHashMap<>();
        stats.put("totalConnections", SSE_CACHE.size());
        stats.put("activeConnections", SSE_CACHE.size());
        return stats;
    }
    public  Map<String, TopologyPointDataVO> getTopologMap() {
    	return TOPOLOGYMAP;
    }
    public void setTopologMap(String nodeKey,TopologyPointDataVO topologyPointData) {
    	 TOPOLOGYMAP.put(nodeKey, topologyPointData);
    }
}